package com.ccubee.data.transfer.transfer.mod;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.otter.canal.protocol.CanalEntry;

import com.ccubee.data.transfer.event.Cmd;
import com.ccubee.data.transfer.event.ColumnData;
import com.ccubee.data.transfer.event.EventTypeContext;
import com.ccubee.data.transfer.event.SqlData;
import com.ccubee.data.transfer.jdbc.JdbcPoolUtil;
import com.ccubee.data.transfer.model.TransferRequest;
import com.ccubee.data.transfer.properties.DataProperties;
import com.ccubee.data.transfer.properties.EventTypeProperties;
import com.ccubee.data.transfer.properties.transfer.DbProperties;
import com.ccubee.data.transfer.transfer.ICall;
import com.ccubee.data.transfer.transfer.monitor.DataTransferMonitor;
import com.ccubee.data.transfer.transfer.monitor.MonitorProperties;
import lombok.Getter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import javax.sql.DataSource;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 抽象的模式
 *
 * @Author ycf
 * @Date 2022/11/15 13:20
 * @Version 1.0
 */
@Slf4j
@Getter
public abstract class AbstractMod implements Imod, ICall, ICommonMod {
    /**
     * 当前请求参数
     */
    protected TransferRequest transferRequest;
    /**
     * 数据库配置
     */
    protected List<DbProperties> dbProperties;
    /**
     * 线程池
     */
    protected ExecutorService executor;
    /**
     * 数据属性
     */
    protected DataProperties dataProperties;
    /**
     * 实例名称
     */
    protected String instance;
    /**
     * 定时任务线程池
     */
    protected ScheduledExecutorService executorService;

    @Override
    public Imod init(String instance, List<DbProperties> dbProperties, DataProperties dataProperties, ExecutorService executor, ScheduledExecutorService executorService) {
        this.instance = instance;
        this.dbProperties = dbProperties;
        this.executor = executor;
        this.dataProperties = dataProperties;
        this.executorService = executorService;
        return this;
    }

    @Override
    public ICall exe(TransferRequest transferRequest) {
        this.transferRequest = transferRequest;
        return this;
    }

    /**
     * 获取sql数据
     *
     * @param dbName    数据库名称
     * @param tableName 表名
     * @param num       当前行
     * @return 结果
     */
    protected abstract SqlData sqlData(String dbName, String tableName, long num);

    @Override
    public void start() {
        dbProperties.forEach(i -> table(dataProperties.dataSource(i.getDsSource()), i));
    }

    @Override
    public void stop() {
        log.info("job stop {}", getInstance());
        DataTransferMonitor.monitor(getInstance())
                .forEach((k, v) -> v.forEach((k1, v1) -> v1.setRun(false)));
        DataTransferMonitor.monitor(getInstance())
                .forEach((k, v) -> v.forEach((k1, v1) -> System.out.println(v1)));
    }

    @Override
    public boolean isRun() {
        return DataTransferMonitor.monitor(getInstance())
                .entrySet()
                .stream()
                .flatMap(i -> i.getValue().entrySet().stream())
                .anyMatch(i -> i.getValue().getRun());
    }

    /**
     * 结束当前参数任务
     *
     * @param i            表配置
     * @param dbProperties 数据库配置
     */
    public void stop(DbProperties.TableConfig i, DbProperties dbProperties) {
        monitor(dbProperties, i, false);
    }

    /**
     * 当前参数获取是否开始
     *
     * @param i            表配置
     * @param dbProperties 数据库配置
     * @return 是否运行状态
     */
    protected boolean isStart(DbProperties.TableConfig i, DbProperties dbProperties) {
        return DataTransferMonitor.monitorGet(getInstance(),
                formatKey(dbProperties.getSource(),
                        dbProperties.getTarget()),
                formatKey(i.getSource(),
                        i.getTarget())).getRun();
    }

    /**
     * 获取计数
     *
     * @param maps 数据
     * @param num  当前数
     * @return 结果
     */
    protected abstract long num(List<List<ColumnData>> maps, long num);

    /**
     * 数据处理
     *
     * @param dataSource   数据源
     * @param dbProperties 数据库配置
     */
    private void table(DataSource dataSource, DbProperties dbProperties) {
        dbProperties.getTables()
                .forEach(i -> {
                    MonitorProperties monitor = monitor(dbProperties, i);
                    monitor.setRun(true);
                    log.info("start task:{}->{},data:{}", dbProperties.getSource(), dbProperties.getTarget(), monitor);
                    executor.execute(() -> task(dataSource, i, dbProperties, EventTypeProperties.builder().insertOrUpdate(i.getInsertOrUpdate()).build(), monitor.getStart()));
                });
    }

    /**
     * 任务
     *
     * @param dataSource   数据源
     * @param i            表配置
     * @param dbProperties 数据库配置
     * @param build        信息
     * @param num          当前
     */
    @SneakyThrows
    private void task(DataSource dataSource, DbProperties.TableConfig i, DbProperties dbProperties, EventTypeProperties build, long num) {
        if (log.isDebugEnabled()) {
            log.debug("task :{}->{},{}->{} start num:{}", dbProperties.getSource(), dbProperties.getTarget(), i.getSource(), i.getTarget(), num);
        }
        List<List<ColumnData>> maps = JdbcPoolUtil.executeQuery(dataSource, sqlData(dbProperties.getSource(), i.getSource(), num), i);
        if (CollUtil.isEmpty(maps)) {
            end(i, dbProperties, num);
            return;
        }
        monitor(dbProperties, i, num);
        maps.forEach(i2 -> executor.execute(() -> EventTypeContext.cmd(CanalEntry.EventType.INSERT)
                .cmd(Cmd.builder()
                        .tableName(i.getTarget())
                        .schemaName(dbProperties.getTarget())
                        .destination(instance)
                        .dataSource(dbProperties.getDsTarget())
                        .typeProperties(build)
                        .dataList(i2).build())));
        long l = (maps.size() < transferRequest.getPageSize()) ? -1 : num(maps, num);
        maps.clear();
        loop(dataSource, i, dbProperties, build, l);
    }

    /**
     * 循环处理
     *
     * @param dataSource   数据源
     * @param i            表配置
     * @param dbProperties 数据库配置
     * @param build        事件类型
     * @param num          当前计数
     */
    @SneakyThrows
    private void loop(DataSource dataSource, DbProperties.TableConfig i, DbProperties dbProperties, EventTypeProperties build, long num) {
        if (isEnd(num)) {
            end(i, dbProperties, num);
            return;
        }
        if (!isStart(i, dbProperties)) {
            log.warn("task is end :{}->{},{}->{} end:{}", dbProperties.getSource(), dbProperties.getTarget(), i.getSource(), i.getTarget(), num);
            return;
        }
        executorService.schedule(() -> task(dataSource, i, dbProperties, build, num), transferRequest.getSleep(), TimeUnit.MILLISECONDS);
    }

    private void end(DbProperties.TableConfig i, DbProperties dbProperties, long num) {
        log.warn("task :{}->{},{}->{} end:{}", dbProperties.getSource(), dbProperties.getTarget(), i.getSource(), i.getTarget(), num);
        stop(i, dbProperties);
        monitor(dbProperties, i, Boolean.FALSE, num);
    }

    /**
     * 是否结束
     *
     * @param num 计数
     * @return 结果
     */
    private boolean isEnd(long num) {
        if (num == -1) {
            return true;
        } else {
            return transferRequest.getEnd() > 0 && num >= transferRequest.getEnd();
        }
    }
}
